package com.atguigu.flink.chapter11.window;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2022/11/1 09:34
 */
public class Flink02_Window_TVF {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<WaterSensor> stream = env.fromElements(
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s2", 1000L, 10),
            new WaterSensor("s1", 2000L, 20),
            new WaterSensor("s1", 3000L, 30),
            new WaterSensor("s1", 4000L, 40),
            new WaterSensor("s1", 5000L, 50),
            new WaterSensor("s1", 8000L, 50)
        ).assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((ws, ts) -> ws.getTs())
        );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts"), $("vc"), $("et").rowtime());
        tEnv.createTemporaryView("sensor", table);
        
      
        // tvf: 滚动窗口
        /*tEnv.sqlQuery("select" +
                          " window_start, window_end, id, " +
                          " sum(vc) vc_sum " +
                          "from table( tumble(table sensor, descriptor(et), interval '5' second ) ) " +
                          "group by window_start, window_end, id")  // window_start, window_end 务必要出现在 group by 中
            .execute()
            .print();*/
    
    
        tEnv.sqlQuery("select" +
                          " window_start, window_end, id, " +
                          " sum(vc) vc_sum " +
                          "from table( hop(table sensor, descriptor(et), interval '2' second, interval '6' second ) ) " +
                          "group by window_start, window_end, id")  // window_start, window_end 务必要出现在 group by 中
            .execute()
            .print();
 
        
    }
}
/*
分组窗口(group window)
    keyBy-> window-> agg
    
    0-5  s1  10
    0-5  s2  100
    5-10 s1  20
    ...
    select w.start, w.end id, sum(vc) from t group by id, w
    


over 窗口


 */